package com.greate.community.service;

import com.greate.community.config.KafkaConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.scheduling.annotation.Scheduled;

import javax.annotation.PostConstruct;
import java.util.Collections;
import java.util.List;
import java.util.ArrayList;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Kafka 消费者服务类
 */
@Service
public class Demo {

    private static final Logger logger = LoggerFactory.getLogger(Demo.class);

    @Autowired
    private KafkaConsumerConfig kafkaConsumerConfig;

    // 用来控制消费者的数量
    private AtomicBoolean isExpanding = new AtomicBoolean(false);

    // 当前的消费者列表
    private List<KafkaConsumer<String, String>> commentConsumers;

    // 消费的主题和消费者组
    private static final String TOPIC_COMMNET = "comment";
    private static final String COMMENT_CONSUMER_GROUP = "comment-consumer-group";

    private static final int INITIAL_CONSUMER_COUNT = 3; // 初始消费者数量
    private static final int MAX_CONSUMER_COUNT = 10;    // 最大消费者数量（假设）

    @PostConstruct
    public void init() {
        // 初始化时创建初始的消费者实例
        commentConsumers = createConsumers(COMMENT_CONSUMER_GROUP, INITIAL_CONSUMER_COUNT, TOPIC_COMMNET);

        // 启动每个消费者的线程
        for (KafkaConsumer<String, String> consumer : commentConsumers) {
            new Thread(() -> consumeMessages(consumer, TOPIC_COMMNET)).start();
        }
    }

    /**
     * 创建消费者实例
     *
     * @param groupId      消费者组 ID
     * @param consumerCount 消费者数量
     * @param topic        订阅的主题
     * @return 创建的消费者列表
     */
    private List<KafkaConsumer<String, String>> createConsumers(String groupId, int consumerCount, String topic) {
        List<KafkaConsumer<String, String>> consumers = new ArrayList<>();
        for (int i = 0; i < consumerCount; i++) {
            KafkaConsumer<String, String> consumer = kafkaConsumerConfig.createConsumer(groupId);
            consumer.subscribe(Collections.singletonList(topic));
            consumers.add(consumer);
        }
        return consumers;
    }

    /**
     * 消费消息并处理
     *
     * @param consumer 消费者实例
     * @param topic    订阅的主题
     */
    private void consumeMessages(KafkaConsumer<String, String> consumer, String topic) {
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                for (ConsumerRecord<String, String> record : records) {
                    // 处理消费到的消息
                    logger.info("消费到的消息：{}，来自主题：{}", record.value(), topic);
                }
            }
        } catch (Exception e) {
            logger.error("消费消息时发生错误", e);
        }
    }

    /**
     * 定时每 5 秒检查一次消息
     */
    @Scheduled(fixedRate = 5000)
    public void consumeCommentMessages() {
        logger.info("开始从 Kafka 拉取评论事件消息...");
        // 遍历每个消费者并消费消息
        for (KafkaConsumer<String, String> consumer : commentConsumers) {
            consumeMessages(consumer, TOPIC_COMMNET);
        }
    }

    /**
     * 动态扩展消费者数量
     *
     * @param increaseBy 增加的消费者数量
     */
    public synchronized void expandConsumers(int increaseBy) {
        if (increaseBy <= 0 || isExpanding.get()) {
            return;  // 防止重复扩展
        }

        // 当前消费者数量
        int currentConsumerCount = commentConsumers.size();
        int newConsumerCount = Math.min(currentConsumerCount + increaseBy, MAX_CONSUMER_COUNT);

        if (currentConsumerCount < newConsumerCount) {
            isExpanding.set(true);
            int consumersToAdd = newConsumerCount - currentConsumerCount;
            List<KafkaConsumer<String, String>> newConsumers = createConsumers(COMMENT_CONSUMER_GROUP, consumersToAdd, TOPIC_COMMNET);
            commentConsumers.addAll(newConsumers);

            // 启动新添加的消费者
            for (KafkaConsumer<String, String> consumer : newConsumers) {
                new Thread(() -> consumeMessages(consumer, TOPIC_COMMNET)).start();
            }

            logger.info("成功扩展了 {} 个消费者，当前总消费者数量：{}", consumersToAdd, newConsumerCount);
            isExpanding.set(false);
        }
    }
}
